package cn.itcast.flink.base;

import lombok.AllArgsConstructor;
import lombok.Data;
import lombok.NoArgsConstructor;
import org.apache.flink.configuration.Configuration;
import org.apache.flink.streaming.api.datastream.DataStreamSource;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.functions.source.RichSourceFunction;

import java.sql.Connection;
import java.sql.DriverManager;
import java.sql.PreparedStatement;
import java.sql.ResultSet;

/**
 * Author itcast
 * Date 2021/7/27 11:13
 * 需求： 读取 t_student表中的数据并加载到数据集中
 * 开发步骤：
 * 1. 获取流执行环境
 * 2. 添加mysql数据源
 * 3. 打印输出
 * 4. 执行流环境
 */
public class MySQLSource {
    public static void main(String[] args) throws Exception {
        //0.创建数据库和创建数据表
        /**
         * create database if not exists bigdata;
         * use bigdata;
         * CREATE TABLE if not exists `t_student` (
         *     `id` int(11) NOT NULL AUTO_INCREMENT,
         *     `name` varchar(255) DEFAULT NULL,
         *     `age` int(11) DEFAULT NULL,
         *     PRIMARY KEY (`id`)
         * ) ENGINE=InnoDB AUTO_INCREMENT=7 DEFAULT CHARSET=utf8;
         *
         * INSERT INTO `t_student` VALUES ('1', 'jack', '18');
         * INSERT INTO `t_student` VALUES ('2', '张三', '19');
         * INSERT INTO `t_student` VALUES ('3', 'rose', '20');
         * INSERT INTO `t_student` VALUES ('4', 'tom', '19');
         * INSERT INTO `t_student` VALUES ('5', '李四', '18');
         * INSERT INTO `t_student` VALUES ('6', '王五', '20');
         *
         * mysql> desc t_student;
         * +-------+--------------+------+-----+---------+----------------+
         * | Field | Type         | Null | Key | Default | Extra          |
         * +-------+--------------+------+-----+---------+----------------+
         * | id    | int(11)      | NO   | PRI | NULL    | auto_increment |
         * | name  | varchar(255) | YES  |     | NULL    |                |
         * | age   | int(11)      | YES  |     | NULL    |                |
         * +-------+--------------+------+-----+---------+----------------+
         */
        //1. 获取流执行环境
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        //2. 添加mysql数据源
        //todo 2.1 RichSourceFunction 增强的读取数据源的函数，可以获取当前的上下文的参数
        DataStreamSource<Student> source = env.addSource(new RichSourceFunction<Student>() {
            boolean flag = true;
            Connection conn = null;
            PreparedStatement ps = null;
            ResultSet rs = null;

            //todo 2.2 连接数据库，读取到表中数据，要在 open 方法中实现
            @Override
            public void open(Configuration parameters) throws Exception {
                conn = DriverManager
                        .getConnection("jdbc:mysql://node3:3306/bigdata?useSSL=false", "root", "123456");
                //生成prepareStatement
                String sql = "SELECT id,`name`,age FROM t_student";
                ps = conn.prepareStatement(sql);
                rs = ps.executeQuery();
            }

            //todo 2.3 读取出来 t_student 中的数据，每2秒钟读取一条数据
            @Override
            public void run(SourceContext<Student> ctx) throws Exception {
                while (flag) {
                    while (rs.next()) {
                        int id = rs.getInt("id");
                        String name = rs.getString("name");
                        int age = rs.getInt("age");
                        ctx.collect(new Student(id, name, age));
                        Thread.sleep(2000L);
                    }
                }
            }

            //todo 2.4 如果用户取消 cancel
            @Override
            public void cancel() {
                flag = false;
            }

            //todo 2.5 当前的数据库读取完之后，关闭连接，关闭数据集
            @Override
            public void close() throws Exception {
                if (!rs.isClosed()) rs.close();
                if (!ps.isClosed()) ps.close();
                if (!conn.isClosed()) conn.close();
            }
        });


        //3. 打印输出
        source.print();
        //4. 执行流环境e
        env.execute();
    }
    @Data
    @AllArgsConstructor
    @NoArgsConstructor
    public static class Student{
        private int id;
        private String name;
        private int age;
    }
}
